package com.shujia.flink.tf

import org.apache.flink.api.common.functions.JoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

object Demo10WindowJoin {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //id和姓名
    val idAndNameDS: DataStream[(String, String)] = env
      .socketTextStream("master", 8888)
      .map(line => {
        val split: Array[String] = line.split(",")
        (split.head, split.last)
      })

    //id和年龄流
    val idAndAgeDS: DataStream[(String, Int)] = env
      .socketTextStream("master", 9999)
      .map(line => {
        val split: Array[String] = line.split(",")
        (split.head, split.last.toInt)
      })

    val joinDS: DataStream[(String, String, Int)] = idAndNameDS
      .join(idAndAgeDS) //join
      .where(_._1) //选择左表的关联字段
      .equalTo(_._1) //选择右表的关联字段
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) //划分窗口，没5秒一个窗口
      .apply(new JoinFunction[(String, String), (String, Int), (String, String, Int)] {
        /**
         * join： 用于整理数据
         *
         * @param first  ： 左表的数据
         * @param second ：右表的数据
         * @return： 返回结果
         */
        override def join(first: (String, String),
                          second: (String, Int)): (String, String, Int) = {
          val (id, name) = first
          val (_, age) = second

          (id, name, age)
        }
      })

    joinDS.print()

    env.execute()
  }

}
